package com.jch.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.util.Iterator;

public class HbaseTest {

    Connection connection;
    HBaseAdmin admin;


    @Before
    public void init() throws Exception{

        //连接zookeeper
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","hadoop101,hadoop102,hadoop103");

        //连接
        connection = ConnectionFactory.createConnection(conf);

        admin=(HBaseAdmin) this.connection.getAdmin();

    }


    @Test
    public void create() throws Exception{

        //创建表
        TableDescriptorBuilder user_info = TableDescriptorBuilder.newBuilder(TableName.valueOf("user_info"));
        //列族
        ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
        ColumnFamilyDescriptor data = ColumnFamilyDescriptorBuilder.of("data");

        //将表名和列族连接起来
        user_info.setColumnFamily(info);
        user_info.setColumnFamily(data);

        //构造表
        TableDescriptor build = user_info.build();
        admin.createTable(build);
    }

    //添加数据
    @Test
    public void put() throws Exception{
        //获取表名
        Table user_info = connection.getTable(TableName.valueOf("user_info"));
        Put put1 = new Put(Bytes.toBytes("rk0001"));
        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),
                Bytes.toBytes("jch"));
        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),
                Bytes.toBytes("18"));
        Put put2 = new Put(Bytes.toBytes("rk0002"));
        put2.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),
                Bytes.toBytes("zmy"));
        put2.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),
                Bytes.toBytes("19"));

        Put put3 = new Put(Bytes.toBytes("rk0003"));
        put3.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),
                Bytes.toBytes("ym"));
        put3.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),
                Bytes.toBytes("19"));

        user_info.put(put1);
        user_info.put(put2);
        user_info.put(put3);
    }


    //获取表数据
    @Test
    public void show() throws Exception{

        //获取表名
        Table user_info = connection.getTable(TableName.valueOf("user_info"));
        Scan scan = new Scan();

        ResultScanner scanner = user_info.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        while (iterator.hasNext()){

            Result next = iterator.next();
            Cell[] cells = next.rawCells();

            for (Cell cell:cells){
                System.out.println("行键"+Bytes.toString(CellUtil.cloneRow(cell))+"\t"+"列族"+Bytes.toString(CellUtil.cloneFamily(cell))
                +"\t"+"列名"+Bytes.toString(CellUtil.cloneQualifier(cell))+"\t"+"值"+Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

    }




}
